package com.tianqu.tms.core.webservice.weichat.api.mq;

import java.net.ConnectException;
import java.util.ArrayList;
import java.util.Properties;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.tianqu.tms.core.base.Logger;
import com.tianqu.tms.core.webservice.weichat.api.Listener;

import net.chedaiban.wechat.api.mq.protobuf.Rpc.RpcMessage;
/**
 */
public class TransferMqConsumer {
	private Logger LOGGER = Logger.getLogger(TransferMqConsumer.class);
	protected final ArrayList<Listener> listeners = new ArrayList(128);
    static Consumer consumer = null;
    private static TransferMqConsumer INSTANCE = new TransferMqConsumer();

    private TransferMqConsumer() {
    };

    public static TransferMqConsumer getInstance() {
        return INSTANCE;
    }

    public void start() throws ConnectException {
        try {
            Properties consumerProperties = new Properties();
            consumerProperties.setProperty(PropertyKeyConst.ConsumerId, "CID_TMSV2_"+MqConfig.TRANSFER_MQ_APP_TO_TMS_TAG);
            consumerProperties.setProperty(PropertyKeyConst.AccessKey, MqConfig.ACCESS_KEY);
            consumerProperties.setProperty(PropertyKeyConst.SecretKey, MqConfig.SECRET_KEY);
            consumerProperties.setProperty(PropertyKeyConst.ONSAddr, MqConfig.ADDRESS);
            consumer = ONSFactory.createConsumer(consumerProperties);
            consumer.subscribe(MqConfig.TRANSFER_MQ_APP_TO_TMS_TOPIC, MqConfig.TRANSFER_MQ_APP_TO_TMS_TAG, new MessageListenerImpl());
            consumer.start();
        } catch (Exception e) {
            throw new ConnectException("failted to connect aliyun mq!network is error!");
        }

    }

    class MessageListenerImpl implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext consumeContext) {
            try {
            	RpcMessage rpcMessage = RpcMessage.parseFrom(message.getBody());
            	for (Listener listener : listeners)  listener.onMessage(message,rpcMessage);
            }catch (Exception e) {
            	LOGGER.error(e);
			} 
            return Action.CommitMessage;
        }
    }

	public ArrayList<Listener> getListeners() {
		return listeners;
	}


	public boolean addListener(Listener listener) {
		return listeners.add(listener);
	}

	public boolean delListener(Listener listener) {
		return listeners.remove(listener);
	}
    
	//app to tms
	public static void main(String[] args) {
		TransferMqConsumer.getInstance();
	}
}
